package com.sjk.flink.kafka;
import com.sjk.flink.ExpireStreamProcess;
import com.sjk.flink.functions.JsonValidationProcessFunction;
import com.sjk.flink.util.ObjectMapperUtil;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Base64;

/**
 * @author: z
 */
public class MyKafkaDeserializationSchema implements KafkaDeserializationSchema<ConsumerRecord<String, String>> {

    private static final Logger log = LoggerFactory.getLogger(ExpireStreamProcess.class);

    private static  String encoding = "UTF8";
    @Override
    public boolean isEndOfStream(ConsumerRecord<String, String> nextElement) {
        return false;
    }
    @Override
    public ConsumerRecord<String, String> deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
        //System.out.println("Record--partition::"+record.partition());
        //System.out.println("Record--offset::"+record.offset());
        //System.out.println("Record--timestamp::"+record.timestamp());
        //System.out.println("ALL ConsumerRecord::"+ ObjectMapperUtil.getObjectMapper().writeValueAsString(record));
        //System.out.println("Record--timestampType::"+record.timestampType());
        //System.out.println("Record--checksum::"+record.checksum());
        //System.out.println("Record--key::"+record.key());
        //System.out.println("Record--value::"+record.value());
        //log.info("Record--value::                    "+record.value());
        String key = null;
        if(record.key()!=null){
            key = new String((byte[])record.key(), encoding);
        }
        String value = null;
        if(record.value()!=null){
            //log.warn("Record--value::                    "+record.value());
            value =  new String((byte[])record.value(), encoding);
        }
        return new ConsumerRecord(record.topic(),
                record.partition(),
                record.offset(),
                record.timestamp(),
                record.timestampType(),
                record.checksum(),
                record.serializedKeySize(),
                record.serializedValueSize(),
                key,
                value);
    }
    @Override
    public TypeInformation<ConsumerRecord<String, String>> getProducedType() {
        return TypeInformation.of(new TypeHint<ConsumerRecord<String, String>>(){});
    }
}